python

2023-04-15 12:27| 来源: 网络整理| 查看: 265

python | 同时使用flask和websockets

一个非常简单的解决方案

代码如下：

''' author: Mz1 一个脚本启动的多功能聊天室架构: 在新线程中启动flask 在主线程中启动websockets ''' # index.html index_html = ''' PyChatRoom@Mz1 当前聊天室: {{ _path }} /** 0：未连接 1：连接成功，可通讯 2：正在关闭 3：连接已关闭或无法打开 */ //创建一个webSocket 实例 var _url = "ws://127.0.0.1:2333"; var webSocket = new WebSocket(_url); webSocket.onerror = function (event){ onError(event); }; // 打开websocket webSocket.onopen = function (event){ onOpen(event); }; //监听消息 webSocket.onmessage = function (event){ onMessage(event); }; webSocket.onclose = function (event){ onClose(event); } //关闭监听websocket function onError(event){ document.getElementById("msg").innerHTML = "

"; console.log("error"+event.data); }; function onOpen(event){ console.log("open:"+sockState()); document.getElementById("msg").innerHTML = "

Connect to Service

"; document.getElementById("msg").innerHTML += '

连接到: '+_url+'

'; }; function onMessage(event){ console.log("onMessage"); document.getElementById("msg").innerHTML += "

response: "+event.data+"

" }; function onClose(event){ document.getElementById("msg").innerHTML = "

"; console.log("close: "+sockState()); webSocket.close(); } function sockState(){ var status = ['未连接','连接成功，可通讯','正在关闭','连接已关闭或无法打开']; return status[webSocket.readyState]; } function start(event){ console.log(webSocket); var msg = document.getElementById('text').value; document.getElementById('text').value = ''; console.log("send:"+sockState()); console.log("msg="+msg); webSocket.send("msg="+msg); document.getElementById("msg").innerHTML += "

request"+msg+"

" }; function close(event){ webSocket.close(); } ''' from flask import Flask, render_template_string import asyncio import websockets import threading import time # Websocket服务端 class WebsocketChatServer(): def __init__(self): pass async def run(self, port): start_server = websockets.serve(self.handler, "", port) await start_server print(f' > server start ok! on port {port}') await asyncio.Future() # run forever async def handler(self, websocket, path): print(path) while True: try: msg = await websocket.recv() except websockets.ConnectionClosedOK: break print(f"recv: {msg}") print(' > close a connection') # 初始化和定义flask app = Flask(__name__) @app.route('/', defaults={'path': ''}) @app.route('/') def index(path): if path == '': path = 'home' # 默认聊天室 return render_template_string(index_html, _path=path) # 捕获一切路径，返回拼接的html文件 # 启动函数 def main(): # 在新线程中启动flask print(app.url_map) flask_thread = threading.Thread(target=app.run, args=('0.0.0.0', 5000)) flask_thread.start() # 启动websocket server print('> starting server...') server = WebsocketChatServer() tasks = [ server.run(2333), ] loop = asyncio.get_event_loop() try: loop.run_until_complete(asyncio.wait(tasks)) except KeyboardInterrupt: for task in asyncio.Task.all_tasks(): task.cancel() loop.stop() loop.run_forever() loop.close() if __name__ == '__main__': main()

本文来自博客园，作者：Mz1，转载请注明原文链接：https://www.cnblogs.com/Mz1-rc/p/16774938.html

如果有问题可以在下方评论或者email:[email protected]

【本文地址】

公司简介

联系我们